package org.apache.flume.sink.hbase.kdyzm;

import com.google.common.base.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.sink.hbase.AsyncHbaseEventSerializer;
import org.apache.flume.sink.hbase.SimpleHbaseEventSerializer;
import org.apache.flume.sink.hbase.SimpleRowKeyGenerator;
import org.hbase.async.AtomicIncrementRequest;
import org.hbase.async.PutRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * @author kdyzm
 * @date 2022/10/2
 */
public class DjtAsyncHbaseEventSerializer implements AsyncHbaseEventSerializer {

    private byte[] table;
    private byte[] cf;
    private byte[] payload;
    private byte[] payloadColumn;
    private byte[] incrementColumn;
    private String rowPrefix;
    private byte[] incrementRow;
    private SimpleHbaseEventSerializer.KeyType keyType;

    private Logger logger = LoggerFactory.getLogger(DjtAsyncHbaseEventSerializer.class);

    @Override
    public void initialize(byte[] table, byte[] cf) {
        logger.info("DjtAsyncHbaseEventSerializer initialize");
        this.table = table;
        this.cf = cf;
    }

    @Override
    public void setEvent(Event event) {
        logger.info("DjtAsyncHbaseEventSerializer setEvent");
        //获取每条完整日志数据
        this.payload = event.getBody();
    }

    @Override
    public List<PutRequest> getActions() {
        List<PutRequest> actions = new ArrayList<>();
        if (payloadColumn != null) {
            byte[] rowKey;
            try {
                //解析列字段
                String[] columns = new String(this.payloadColumn).split(",");
                //解析每列对应的值
                String[] values = new String(this.payload).split(",");
                //判断数据与字段长度是否一致
                if (columns.length != values.length) {
                    return actions;
                }
                //时间
                String datetime = values[0];
                //用户id
                String userId = values[1];
                //自定义生成RowKey
                rowKey = SimpleRowKeyGenerator.getDjtRowKey(userId, datetime);
                for (int i = 0; i < columns.length; i++) {
                    byte[] colColumn = columns[i].getBytes();
                    byte[] colValue = values[i].getBytes(StandardCharsets.UTF_8);
                    //构造写请求对象
                    PutRequest putRequest = new PutRequest(table, rowKey, cf, colColumn, colValue);
                    actions.add(putRequest);
                }
            } catch (Exception e) {
                throw new FlumeException("Could not get row key!", e);
            }
        }
        logger.info("DjtAsyncHbaseEventSerializer getActions size = {}", actions.size());
        return actions;
    }

    @Override
    public List<AtomicIncrementRequest> getIncrements() {
        List<AtomicIncrementRequest> actions = new ArrayList<>();
        if (incrementColumn != null) {
            AtomicIncrementRequest inc = new AtomicIncrementRequest(table,
                    incrementRow, cf, incrementColumn);
            actions.add(inc);
        }
        return actions;
    }

    @Override
    public void cleanUp() {
        // TODO Auto-generated method stub
    }

    @Override
    public void configure(Context context) {
        //获取后续Flume聚合配置中传入的数据对应的列名称
        String pCol = context.getString("payloadColumn", "pCol");
        String iCol = context.getString("incrementColumn", "iCol");
        rowPrefix = context.getString("rowPrefix", "default");
        String suffix = context.getString("suffix", "uuid");
        if (pCol != null && !pCol.isEmpty()) {
            if (suffix.equals("timestamp")) {
                keyType = SimpleHbaseEventSerializer.KeyType.TS;
            } else if (suffix.equals("random")) {
                keyType = SimpleHbaseEventSerializer.KeyType.RANDOM;
            } else if (suffix.equals("nano")) {
                keyType = SimpleHbaseEventSerializer.KeyType.TSNANO;
            } else {
                keyType = SimpleHbaseEventSerializer.KeyType.UUID;
            }
            payloadColumn = pCol.getBytes(Charsets.UTF_8);
        }
        if (iCol != null && !iCol.isEmpty()) {
            incrementColumn = iCol.getBytes(Charsets.UTF_8);
        }
        incrementRow = context.getString("incrementRow", "incRow").getBytes(Charsets.UTF_8);
    }

    @Override
    public void configure(ComponentConfiguration componentConfiguration) {
        // TODO Auto-generated method stub
    }
}
